package cn.gwm.flink.streaming.function.source;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author : GW00280745
 * @date : 2023/1/15 15:28
 * @description :
 */
public class ChargePredictMysqlSourceFunction extends RichSourceFunction<JSONObject> {
    public ChargePredictMysqlSourceFunction(String url,String user,String password,String sql,String timeout){
        this.url = url;
        this.user = user;
        this.password = password;
        this.sql = sql;
        this.timeout = timeout;
    }
    private String timeout;
    private String url;
    String sql;
    private String user;
    private String password;
    private Connection conn = null;
    private PreparedStatement statement = null;
    private ResultSet resultSet = null;

    private boolean isTest = true;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        try {
            if (isTest) {
                return;
            }
            //连接数据库
            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection(url, user, password);
            statement = conn.prepareStatement(sql);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run(SourceContext<JSONObject> ctx) throws Exception {
        try {
            if (isTest) {
                return;
            }
            while (flag) {
                JSONObject jsonObject = new JSONObject();
                resultSet = statement.executeQuery();
                while (resultSet.next()) {
                    String configKey = resultSet.getString("config_key");
                    Object configValue = resultSet.getObject("config_value");
                    jsonObject.put(configKey, configValue);
                }
                ctx.collect(jsonObject);
                //定时执行
//            COUNT_DOWN_LATCH.await(Long.parseLong(timeout), TimeUnit.MINUTES)
                Thread.sleep(Long.parseLong(timeout));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private boolean flag = true;
//    private CountDownLatch COUNT_DOWN_LATCH = new CountDownLatch(1)  非序列化因此需要转一下才能用

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        if(conn != null){
            conn.close();
        }
        if(statement != null) {
            statement.close();
        }
        if(resultSet != null){
            resultSet.close();
        }
    }
}
